货轮|水位_Flink之Watermark滑动窗口案例

1、丢弃&＃xff08;默认的处理方式&＃xff09;
在Flink当中&＃xff0c;如果输入数据所在的窗口已执行过了&＃xff0c;Flink对这些延迟数据的处理方案默认就是丢弃&＃xff0c;而不是再次出发响应的window窗口。
2、allowedLateness指定允许数据延迟的时间
在Flink当中&＃xff0c;当输入数据所在的窗口已经执行过了&＃xff0c;默认情况下即使再来心的数据&＃xff0c;window也不会再次出发&＃xff0c;但是如果我们希望再次被触发咋么解决&＃xff1f;
即在某些情况下&＃xff0c;我们希望为延迟的数据提供一个宽容的时间。

Flink提供了allowwedLateness方法&＃xff0c;它可以实现对延迟数据设置一个延迟的时间&＃xff0c;在指定延迟时间内到达的数据可以被再次出发窗口window计算。
在这里我们可以用一个列子来说明问题&＃xff1a;
maxOutOfOrderness表示允许数据的最大乱序时间&＃xff1a;好比我们的大货轮10:00开船&＃xff0c;但是大货轮给乘客提供了5分钟的延迟时间&＃xff0c;10:05开船&＃xff1b;
allowedLateness表示是否可以再次触发窗口的延迟时间&＃xff1a;好比大货轮10:05已经开船&＃xff0c;但是大货轮又给乘客提供了2分钟的延迟时间&＃xff0c;即只要大货轮在2min内的触发时间&＃xff0c;都可以给你提供一个梯子&＃xff0c;让你再次爬山来。

waterMark允许数据延迟时间与这个数据延迟的区别是;allowedLateness允许延迟时间在Watermark允许延迟时间的基础上增加的时间。

所谓的延迟数据&＃xff0c;即窗口已经因为watermark进行了触发&＃xff0c;则在此之后如果还有数据进入窗口&＃xff0c;则默认情况下不会对窗口window进行再次触发和聚合计算。要想在数据进入已经被触发过的窗口后&＃xff0c;还能继续触发窗口计算&＃xff0c;则可以使用延迟数据处理机制。

因此如果你加上了allowedLateness参数&＃xff0c;窗口的触发条件是&＃xff1a;
a、窗口第一次触发是在Watermark时间>&＃61;window中的window_end_time;
b、第二次&＃xff08;或多次&＃xff09;触发的条件是Watermark时间

3、sideOutputLateDate手机延迟数据
通过sideOutputLatedate函数可以把延迟的数据统一收集&＃xff0c;统一存储&＃xff0c;方便后续排查问题。

 处理延迟数据&＃xff1a;

package Flink_Window;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
//watermark滑动窗口案例
public class SocketCountWindowEvent
public static void main(String[] args) throws Exception
StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//连接socket
DataStreamSource dataStreamSource &＃61; env.socketTextStream("192.168.208.121", 8888, "\\n");
DataStream streamOperator &＃61; dataStreamSource.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks()
//设定水印&＃xff0c;水印是只增不减的
//maxOutOfOrderness表示允许数据最大乱序事件是30s
Long maxOutOfOrderness &＃61; 30000L;//30s
Long currentMaxTimestamp &＃61;0L;
&＃64;Nullable
&＃64;Override
public Watermark getCurrentWatermark()
return new Watermark(currentMaxTimestamp-maxOutOfOrderness);

&＃64;Override
public long extractTimestamp(String s, long l)
String[] split &＃61;s.split(",");
long event_time&＃61;Long.parseLong(split[1]);
//水位线只增不减
currentMaxTimestamp&＃61;Math.max(event_time,currentMaxTimestamp);
return event_time;

);
//WordCount程序主逻辑
DataStream> windowCounts &＃61; streamOperator.flatMap(new FlatMapFunction>()
&＃64;Override
public void flatMap(String s, Collector> collector) throws Exception
String[] split &＃61; s.split("\\\\W&＃43;");
collector.collect(Tuple2.of(split[0],1));

);
//保存被丢弃的数据
OutputTag> outputTag&＃61;new OutputTag>("late_date");
//注意&＃xff1a;由于getSideOutput方法是SingleOutputStreamOperator类中特有的方法&＃xff0c;所以这里不能用DataStream.
SingleOutputStreamOperator> result&＃61;windowCounts.keyBy(0)
//滚动改为滑动窗口
.timeWindow(Time.seconds(60))
//保存被丢弃的数据
.sideOutputLateData(outputTag)
.process(new ProcessWindowFunction, Tuple2, Tuple, TimeWindow>()
SimpleDateFormat sdf&＃61;new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
&＃64;Override
public void process(Tuple tuple, Context context, Iterable> v2s, Collector> collector) throws Exception
System.out.print("&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;");
System.out.print("算子是&＃xff1a;"&＃43;Thread.currentThread().getId()&＃43;"窗口范围是:"&＃43;sdf.format(context.window().getStart())&＃43;"\\t"&＃43;sdf.format(context.window().getEnd()));
//查看那个水印出发了计算
System.out.print(context.currentWatermark()&＃43;"\\t"&＃43;sdf.format(context.currentWatermark()));
int sum &＃61; 0;
for(Tuple2 v2:v2s)
sum &＃43;&＃61; 1;

collector.collect(Tuple2.of(tuple.getField(0),sum));
System.out.print("&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;下&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;&＃43;");
);
//将延迟的数据暂时打印到控制台&＃xff0c;实际中可以保存到其他存储介质当中
DataStream> sideOutput&＃61;result.getSideOutput(outputTag);
sideOutput.print();
result.print();
env.execute("SocketCountWindowEvent");